package edu.csl.study.kafka.demo;

import org.apache.kafka.clients.producer.*;

import java.util.Properties;
import java.util.concurrent.ExecutionException;

public class KafkaProducerDemo {

  public static void main(String[] args) throws ExecutionException, InterruptedException {

    //准备配置属性
    Properties props = new Properties();
    //kafka集群地址
    props.put("bootstrap.servers", "centos20:9092,centos21:9092,centos22:9092");
    //acks它代表消息确认机制
    //1：默认值，leader分区写入成功，不管follow分区是否希尔
    //0：只管写，不管有没有成功。
    //-1或all：所有的分区都确认才响应（最安全）
    props.put("acks", "all");
    //写入失败后的重试的次数
    props.put("retries", 0);
    //批量写入大小，达到16KB才写入
    props.put("batch.size", 16384);
    //如果没有达到16KB，那么延迟1ms也会写入
    props.put("linger.ms", 1);
    //用来批量写入的缓冲区的大小 32M
    props.put("buffer.memory", 33554432);
    props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
    props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

    //在Properties对象中添加自定义分区类
    props.put("partitioner.class",CustomerPartitioner.class.getName());
    //拦截器
    props.put("interceptor.classes",ProducerInterceptorPrefix.class.getName());

    String topic = "order";

    Producer<String, String> producer = new KafkaProducer<String, String>(props);
    for (int i = 0; i < 5; i++) {
      //这里需要三个参数，第一个：topic的名称，第二个参数：表示消息的key,第三个参数：消息具体内容
      //1、不给定具体的分区号，给定一个key值, 这里使用key的 hashcode%分区数=分区号
      //producer.send(new ProducerRecord<String, String>(topic, Integer.toString(i), "hello-kafka-" + i));

      //2、给定具体的分区号，数据就会写入到指定的分区中
      //producer.send(new ProducerRecord<String, String>(topic, 0,Integer.toString(i), "hello-kafka-"+i));

      //3、不给定具体的分区号，也不给定对应的key ,这个它会进行轮训的方式把数据写入到不同分区中
      //producer.send(new ProducerRecord<String, String>(topic, "hello-kafka-"+i));

      //4、自定义分区
      //异步方式发送消息
      producer.send(new ProducerRecord<String, String>(topic, 0, Integer.toString(i), "hello-kafka-" + i), new Callback() {
        public void onCompletion(RecordMetadata metadata, Exception exception) {
            if(exception  == null){
              System.out.println("消息发送成功！");
            }else{
              System.out.println("消息发送失败！");
            }
        }
      });
      //同步方式：吞吐量不如异步方式。
      //producer.send(xxx).get()

    }
    producer.close();

  }



}
